package io.kiki.sba.registry.server.metadata.remoting.handler;

import io.kiki.sba.registry.api.Channel;
import io.kiki.sba.registry.common.model.GenericResponse;
import io.kiki.sba.registry.common.model.Node;
import io.kiki.sba.registry.common.model.metaserver.cluster.VersionedList;
import io.kiki.sba.registry.common.model.metaserver.inter.heartbeat.HeartBeatResponse;
import io.kiki.sba.registry.common.model.metaserver.inter.heartbeat.HeartbeatRequest;
import io.kiki.sba.registry.common.model.metaserver.nodes.CentralControlNode;
import io.kiki.sba.registry.common.model.metaserver.nodes.ClientInterfaceNode;
import io.kiki.sba.registry.common.model.metaserver.nodes.DataStoreNode;
import io.kiki.sba.registry.common.model.multi.cluster.DataCenter;
import io.kiki.sba.registry.common.model.multi.cluster.RemoteSlotsStatus;
import io.kiki.sba.registry.common.model.slot.Slots;
import io.kiki.sba.registry.exception.MetaLeaderNotWarmupException;
import io.kiki.sba.registry.exception.SofaRegistryMetaLeaderException;
import io.kiki.sba.registry.server.metadata.MetadataLeaderService;
import io.kiki.sba.registry.server.metadata.bootstrap.config.NodeConfig;
import io.kiki.sba.registry.server.metadata.metadataServer.impl.DefaultCurrentNodeClusterMetadataServer;
import io.kiki.sba.registry.server.metadata.monitor.dataStoreNode.DataMessageListener;
import io.kiki.sba.registry.server.metadata.monitor.session.ClientInterfaceNodeMessageListener;
import io.kiki.sba.registry.server.metadata.multi.cluster.MultiClusterSlotsSyncer;
import io.kiki.sba.registry.server.metadata.multi.cluster.MultiClusterSlotsSyncerImpl.RemoteClusterSlotState;
import io.kiki.sba.registry.server.shared.slot.SlotsUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;

import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;

/**
 * Handle session/data node's heartbeat request
 */
public class HeartbeatRequestHandler extends BaseMetaServerHandler<HeartbeatRequest<Node>> {


    private static final Logger logger = LoggerFactory.getLogger(HeartbeatRequestHandler.class);

    @Autowired
    private DefaultCurrentNodeClusterMetadataServer defaultCurrentNodeClusterCentralControlServer;

    @Autowired
    private MetadataLeaderService metadataLeaderService;

    @Autowired(required = false)
    private List<DataMessageListener> dataMessageListeners;

    @Autowired(required = false)
    private List<ClientInterfaceNodeMessageListener> clientInterfaceNodeMessageListeners;

    @Autowired
    private NodeConfig nodeConfig;

    @Autowired
    private MultiClusterSlotsSyncer multiClusterSlotsSyncer;


    @Override
    public Object doHandle(Channel channel, HeartbeatRequest<Node> heartbeat) {
        boolean success = false;
        final Node renewNode = heartbeat.getNode();
        try {
            onHeartbeat(channel, heartbeat);

            Slots slots = defaultCurrentNodeClusterCentralControlServer.getSlots();
            if (!SlotsUtils.isValidSlots(slots)) {
                return new GenericResponse<HeartBeatResponse>().fillFailed("slot-table not valid, check meta-server log for detail");
            }
            HeartBeatResponse response = null;

            final VersionedList<CentralControlNode> centralControlServerClusterVersionedList = defaultCurrentNodeClusterCentralControlServer.getCluster();
            final VersionedList<ClientInterfaceNode> clientInterfaceNodeVersionedList = defaultCurrentNodeClusterCentralControlServer.getClientInterfaceNodeManager().getSessionServerMetaInfo();

            switch (renewNode.getNodeType()) {
            case client_interface:
            case data_store:
                Map<String, RemoteSlotsStatus> remoteSlotsStatus = calculateStatus(heartbeat);
                response = new HeartBeatResponse(true, centralControlServerClusterVersionedList, slots, clientInterfaceNodeVersionedList, metadataLeaderService.getLeader(), metadataLeaderService.getLeaderEpoch(), remoteSlotsStatus);
                break;
            case central_control:
                response = new HeartBeatResponse(true, centralControlServerClusterVersionedList, slots, metadataLeaderService.getLeader(), metadataLeaderService.getLeaderEpoch());
                break;
            default:
                break;
            }
            success = true;
            return new GenericResponse<HeartBeatResponse>().fillSucceed(response);
        } catch (Throwable e) {
            if (e instanceof SofaRegistryMetaLeaderException) {
                SofaRegistryMetaLeaderException exception = (SofaRegistryMetaLeaderException) e;
                HeartBeatResponse response = new HeartBeatResponse(false, exception.getLeader(), exception.getEpoch());
                return new GenericResponse<HeartBeatResponse>().fillFailData(response);
            }

            if (e instanceof MetaLeaderNotWarmupException) {
                MetaLeaderNotWarmupException exception = (MetaLeaderNotWarmupException) e;
                HeartBeatResponse response = new HeartBeatResponse(false, exception.getLeader(), exception.getEpoch());
                return new GenericResponse<HeartBeatResponse>().fillFailData(response);
            }

            logger.error("Node {} renew error!", renewNode, e);
            return new GenericResponse<HeartBeatResponse>().fillFailed("Node " + renewNode + "renew error!");
        } finally {
            logger.info("{},{},addr={}", success ? 'Y' : 'N', renewNode.getNodeType(), channel.getRemoteAddress());
        }
    }

    @SuppressWarnings("unchecked")
    private void onHeartbeat(Channel channel, HeartbeatRequest heartbeat) {
        new DefaultHeartbeatListener(nodeConfig.getLocalDataCenter(), channel).onHeartbeat(heartbeat);
        Node node = heartbeat.getNode();
        switch (node.getNodeType()) {
        case client_interface:
            defaultCurrentNodeClusterCentralControlServer.getClientInterfaceNodeManager().renew((ClientInterfaceNode) node, heartbeat.getDuration());
            onSessionHeartbeat(heartbeat);
            return;
        case data_store:
            defaultCurrentNodeClusterCentralControlServer.getDataStoreNodeManager().renew((DataStoreNode) node, heartbeat.getDuration());
            onDataHeartbeat(heartbeat);
            return;
        case central_control:
            defaultCurrentNodeClusterCentralControlServer.renew((CentralControlNode) node);
            return;
        default:
            break;
        }
        throw new IllegalArgumentException("node type not correct: " + node.getNodeType());
    }

    private void onSessionHeartbeat(HeartbeatRequest<ClientInterfaceNode> heartbeat) {
        if (clientInterfaceNodeMessageListeners == null || clientInterfaceNodeMessageListeners.isEmpty()) {
            return;
        }
        clientInterfaceNodeMessageListeners.forEach(listener -> {
            try {
                listener.onHeartbeat(heartbeat);
            } catch (Throwable th) {
                logger.error("[onDataHeartbeat]", th);
            }
        });
    }

    private void onDataHeartbeat(HeartbeatRequest<DataStoreNode> heartbeat) {
        if (dataMessageListeners == null || dataMessageListeners.isEmpty()) {
            return;
        }
        dataMessageListeners.forEach(listener -> {
            try {
                listener.onHeartbeat(heartbeat);
            } catch (Throwable th) {
                logger.error("[onDataHeartbeat]", th);
            }
        });
    }


    @Override
    public Class interest() {
        return HeartbeatRequest.class;
    }

    /**
     * calculate remoteSlotsStatus
     */
    private Map<String, RemoteSlotsStatus> calculateStatus(HeartbeatRequest<Node> heartbeat) {
        Map<String, RemoteClusterSlotState> dataCenterToRemoteClusterSlotStateMap = multiClusterSlotsSyncer.getDataCenterToRemoteClusterSlotStateMap();
        Map<String, Long> requestSlots = heartbeat.getRemoteClusterslotsEpoch();

        Map<String, RemoteSlotsStatus> map = new HashMap<>();
        for (Entry<String, RemoteClusterSlotState> entry : dataCenterToRemoteClusterSlotStateMap.entrySet()) {
            String remoteDataCenter = entry.getKey();
            RemoteClusterSlotState remoteClusterSlotState = entry.getValue();
            Long requestslotsEpoch = requestSlots.get(remoteDataCenter);
            Slots exist = remoteClusterSlotState.getSlots();
            DataCenter dataCenter = entry.getValue().getDataCenterMetadata();
            if (requestslotsEpoch == null || requestslotsEpoch < exist.getEpoch()) {
                logger.info("[calculateStatus]node:{}, heartbeat request:{}/{}, newslotsEpoch:{}/{}, Slots upgrade: {}", heartbeat.getNode(), remoteDataCenter, requestslotsEpoch, remoteDataCenter, exist.getEpoch(), exist);
                map.put(remoteDataCenter, RemoteSlotsStatus.upgrade(exist, dataCenter));
            } else
                if (requestslotsEpoch > exist.getEpoch()) {
                    // it should not happen, print error log and return false
                    logger.error("[calculateStatus]node:{}, heartbeat request:{}/{}, newslotsEpoch:{}/{}, heartbeat error.", heartbeat.getNode(), remoteDataCenter, requestslotsEpoch, remoteDataCenter, exist.getEpoch());
                    map.put(remoteDataCenter, RemoteSlotsStatus.conflict(exist));
                } else {
                    map.put(remoteDataCenter, RemoteSlotsStatus.notUpgrade(requestslotsEpoch, dataCenter));
                }
        }
        return map;
    }


}
